import 'dart:convert';
import 'dart:io';
import 'dart:math';

import 'package:event_bus/event_bus.dart';
import 'package:flutter_dva/flutter_dva.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';
import 'package:todo_app/models/getters.dart';
import 'package:todo_app/utils/mqtt/MessageEventBus.dart';
import 'package:todo_app/utils/webrtc/RTCMessageEventBus.dart';

class LocalMqttClient {
  static var client = MqttServerClient('123.249.34.189', 'emqx_${DateTime.now().millisecondsSinceEpoch}_${Random().nextInt(1000).toString().padLeft(4, '0')}');
  static int pongCount = 0; // Pong counter
  static String username = "liuhong";
  static String password = "123456";
  static EventBus eventBus = EventBus();
  
  static connect() async {
    /// Set logging on if needed, defaults to off
    client.logging(on: false);

    /// Set the correct MQTT protocol for mosquito
    client.setProtocolV311();

    /// If you intend to use a keep alive you must set it here otherwise keep alive will be disabled.
    client.keepAlivePeriod = 20;
    // 自动重连
    client.autoReconnect = true;
    // 重连事件
    client.onAutoReconnected = onReconnected;

    /// The connection timeout period can be set if needed, the default is 5 seconds.
    client.connectTimeoutPeriod = 2000; // milliseconds

    /// Add the unsolicited disconnection callback
    client.onDisconnected = onDisconnected;

    /// Add the successful connection callback
    client.onConnected = onConnected;


    /// Add a subscribed callback, there is also an unsubscribed callback if you need it.
    /// You can add these before connection or change them dynamically after connection if
    /// you wish. There is also an onSubscribeFail callback for failed subscriptions, these
    /// can fail either because you have tried to subscribe to an invalid topic or the broker
    /// rejects the subscribe request.
    client.onSubscribed = onSubscribed;

    /// Set a ping received callback if needed, called whenever a ping response(pong) is received
    /// from the broker.
    client.pongCallback = pong;

    /// Create a connection message to use or use the default one. The default one sets the
    /// client identifier, any supplied username/password and clean session,
    /// an f a specific one below.
    final connMess = MqttConnectMessage()
        .withClientIdentifier(client.clientIdentifier)
        .withWillTopic('willtopic') // If you set this you must set a will message
        .withWillMessage('我已经下线了')
        .startClean() // Non persistent session for testing
        .withWillQos(MqttQos.atLeastOnce);
    print('Mosquitto client connecting....');
    client.connectionMessage = connMess;

    /// Connect the client, any errors here are communicated by raising of the appropriate exception. Note
    /// in some circumstances the broker will just disconnect us, see the spec about this, we however will
    /// never send malformed messages.
    try {
      await client.connect(username, password);
    } on NoConnectionException catch (e) {
      // Raised by the client when connection fails.
      print('client exception - $e');
      client.disconnect();
    } on SocketException catch (e) {
      // Raised by the socket layer
      print('socket exception - $e');
      client.disconnect();
    }

    /// Check we are connected
    if (client.connectionStatus!.state == MqttConnectionState.connected) {
      print('Mosquitto client connected');
    } else {
      /// Use status here rather than state if you also want the broker return code.
      print('ERROR Mosquitto client connection failed - disconnecting, status is ${client.connectionStatus}');
      client.disconnect();
      exit(-1);
    }

    /// The client has a change notifier object(see the Observable class) which we then listen to to get
    /// notifications of published updates to each subscribed topic.
    client.updates!.listen((List<MqttReceivedMessage<MqttMessage?>>? c) {
      final recMess = c![0].payload as MqttPublishMessage;
      if(recMess.variableHeader?.topicName == null) return;
      var topicName = recMess.variableHeader!.topicName;
      if(topicName.startsWith("/r")) {
        MessageEventBus.receivedMessage(recMess);
      }
      if(topicName.startsWith("/rtc/r")) {
        RTCMessageEventBus.receivedMessage(recMess);
      }
      if(topicName.startsWith("/pingPong/r")) {
        receivedUserClientMessage(recMess);
      }
    });

    /// If needed you can listen for published messages that have completed the publishing
    /// handshake which is Qos dependant. Any message received on this stream has completed its
    /// publishing handshake with the broker.
    client.published!.listen((MqttPublishMessage publishMessage) {
      var topic = publishMessage.variableHeader!.topicName;
      var messageType = publishMessage.header?.messageType;
      var qos = publishMessage.header!.qos;
      if(messageType == MqttMessageType.publish && qos == MqttQos.atLeastOnce) {
        if(topic.startsWith("/s")) {
          MessageEventBus.publishedMessage(publishMessage);
        }
        if(topic.startsWith("/rtc/s")) {
          MessageEventBus.publishedMessage(publishMessage);
        }
      }
    });
  }

  static subscribe(int userId) {
    var topic = '/r/$userId'; // 聊天消息｜通知
    var rtcTopic = '/rtc/r/$userId'; // 音视频通话消息
    client.subscribe(topic, MqttQos.exactlyOnce);
    client.subscribe(rtcTopic, MqttQos.exactlyOnce);
  }

  /// The subscribed callback
  static void onSubscribed(String topic) {
    print('Subscription confirmed for topic $topic');
  }

  /// The unsolicited disconnect callback
  static void onDisconnected() {
    print('OnDisconnected client callback - Client disconnection');
    if (client.connectionStatus!.disconnectionOrigin ==
        MqttDisconnectionOrigin.solicited) {
      print('OnDisconnected callback is solicited, this is correct');
    } else {
      print(
          'OnDisconnected callback is unsolicited or none, this is incorrect - exiting');
      exit(-1);
    }
  }

  /// The successful connect callback
  static void onConnected() {
    print('OnConnected client callback - Client connection was successful');
  }

    /// The successful reconnect callback
  static void onReconnected() {
    print('OnReconnected client callback - Client reconnection was successful');
  }

  static int publishMessage(String topic, dynamic payload, [MqttQos? qos]) {
    MqttClientPayloadBuilder builder = MqttClientPayloadBuilder();
    var str = payload is String ? payload : jsonEncode(payload.toJson());
    var data = builder.addUTF8String(str);
    if(data.payload != null) {
      return client.publishMessage(topic, qos ?? MqttQos.exactlyOnce, data.payload!);
    }
    return -1;
  }
  /// Pong callback
  static void pong() {
    print('Ping response client callback invoked');
    pongCount++;
    publishUserClientPingPong();
  }

  static void publishUserClientPingPong() {
    var system = Provider().store!.rootState.system;
    if(system != null && system.loginSuccess == true) {
      publishMessage("/pingPong/s", UserClientPingPong(
        uid: system.user!.id!,
        cid: client.clientIdentifier,
        time: DateTime.now().millisecondsSinceEpoch
      ), MqttQos.atMostOnce);
    }
  }

  static void subscribeUserClient(int userId) {
    client.subscribe("/pingPong/r/$userId", MqttQos.atMostOnce);
  }

  static void unsubscribeUserClient(int userId) {
    client.unsubscribe("/pingPong/r/$userId");
  }
  // 接收到UserClient ping pong消息
  static receivedUserClientMessage(MqttPublishMessage recMess) {
    var topic = recMess.variableHeader!.topicName;
    var userId = int.tryParse(topic.replaceAll("/pingPong/r/", ""));
    if(userId == null) return;
    final pt = const Utf8Decoder().convert(recMess.payload.message);
    final json = jsonDecode(pt);
    eventBus.fire(UserClientPingPong.fromJson(json));
  }
}

class UserClientPingPong {
  final int uid;
  final String cid;
  final int time;

  UserClientPingPong({ required this.uid, required this.cid, required this.time } );
  UserClientPingPong.fromJson(Map<String, dynamic> json) :
    uid = json["uid"],
    cid = json["cid"],
    time = json["time"]
  ;

  bool get online {
    return (DateTime.now().difference(DateTime.fromMillisecondsSinceEpoch(time))).inSeconds < 300;
  }

  Map<String, dynamic> toJson() {
    return {
      "uid": uid,
      "cid": cid,
      "time": time
    };
  }
}
